package com.company.hive.monitor;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.MetaStoreEventListener;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.events.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Hive MetaStore Hook - 元数据操作监控
 * 
 * 监控删表、删库、表覆盖等高危操作，并通过飞书发送告警
 */
public class MetadataMonitorHook extends MetaStoreEventListener {
    
    private static final Logger LOG = LoggerFactory.getLogger(MetadataMonitorHook.class);
    
    private final AlertSender alertSender;
    private final AlertRuleManager ruleManager;
    private final ExecutorService executorService;
    
    // 配置项
    private static final String FEISHU_WEBHOOK_KEY = "hive.metastore.monitor.feishu.webhook";
    private static final String ALERT_ENABLED_KEY = "hive.metastore.monitor.alert.enabled";
    private static final String ALERT_ASYNC_KEY = "hive.metastore.monitor.alert.async";
    
    public MetadataMonitorHook(Configuration config) {
        super(config);
        
        LOG.info("初始化Hive元数据监控Hook");
        
        // 检查是否启用告警
        boolean alertEnabled = config.getBoolean(ALERT_ENABLED_KEY, true);
        if (!alertEnabled) {
            LOG.warn("告警功能已禁用");
            this.alertSender = null;
            this.ruleManager = null;
            this.executorService = null;
            return;
        }
        
        // 初始化告警发送器
        String webhookUrl = config.get(FEISHU_WEBHOOK_KEY);
        if (webhookUrl == null || webhookUrl.isEmpty()) {
            LOG.warn("未配置飞书Webhook URL，告警功能将不可用");
            this.alertSender = null;
        } else {
            this.alertSender = new AlertSender(webhookUrl);
        }
        
        // 初始化规则管理器
        this.ruleManager = new AlertRuleManager(config);
        
        // 是否异步发送告警
        boolean asyncAlert = config.getBoolean(ALERT_ASYNC_KEY, true);
        if (asyncAlert) {
            this.executorService = Executors.newFixedThreadPool(4);
            LOG.info("使用异步模式发送告警");
        } else {
            this.executorService = null;
            LOG.info("使用同步模式发送告警");
        }
        
        LOG.info("Hive元数据监控Hook初始化完成");
    }
    
    @Override
    public void onDropTable(DropTableEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        if (alertSender == null || ruleManager == null) {
            return;
        }
        
        try {
            Table table = event.getTable();
            String dbName = table.getDbName();
            String tableName = table.getTableName();
            String owner = table.getOwner();
            
            LOG.info("检测到删表操作: {}.{}", dbName, tableName);
            
            // 检查是否需要告警
            if (!ruleManager.shouldAlertForDropTable(dbName, tableName)) {
                LOG.debug("表 {}.{} 在白名单中，跳过告警", dbName, tableName);
                return;
            }
            
            // 构建告警消息
            AlertMessage alert = AlertMessage.builder()
                .alertType("DROP_TABLE")
                .level(AlertLevel.SEVERE)
                .database(dbName)
                .table(tableName)
                .user(owner)
                .detail("分区数", String.valueOf(table.getPartitionKeys().size()))
                .detail("表类型", table.getTableType())
                .detail("存储位置", table.getSd() != null ? table.getSd().getLocation() : "N/A")
                .build();
            
            // 发送告警
            sendAlert(alert);
            
        } catch (Exception e) {
            LOG.error("处理删表事件异常", e);
        }
    }
    
    @Override
    public void onDropDatabase(DropDatabaseEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        if (alertSender == null || ruleManager == null) {
            return;
        }
        
        try {
            Database database = event.getDatabase();
            String dbName = database.getName();
            
            LOG.info("检测到删库操作: {}", dbName);
            
            // 检查是否需要告警
            if (!ruleManager.shouldAlertForDropDatabase(dbName)) {
                LOG.debug("数据库 {} 在白名单中，跳过告警", dbName);
                return;
            }
            
            // 构建告警消息
            AlertMessage alert = AlertMessage.builder()
                .alertType("DROP_DATABASE")
                .level(AlertLevel.CRITICAL)
                .database(dbName)
                .detail("描述", database.getDescription())
                .detail("位置", database.getLocationUri())
                .build();
            
            // 发送告警
            sendAlert(alert);
            
        } catch (Exception e) {
            LOG.error("处理删库事件异常", e);
        }
    }
    
    @Override
    public void onCreateTable(CreateTableEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        if (alertSender == null || ruleManager == null) {
            return;
        }
        
        try {
            Table table = event.getTable();
            String dbName = table.getDbName();
            String tableName = table.getTableName();
            
            // 检查是否存在最近的删除操作（表覆盖检测）
            if (ruleManager.isRecentlyDeleted(dbName, tableName)) {
                LOG.info("检测到表覆盖操作: {}.{}", dbName, tableName);
                
                AlertMessage alert = AlertMessage.builder()
                    .alertType("TABLE_OVERWRITE")
                    .level(AlertLevel.SEVERE)
                    .database(dbName)
                    .table(tableName)
                    .user(table.getOwner())
                    .detail("操作", "短时间内删除后重建")
                    .build();
                
                sendAlert(alert);
            }
            
        } catch (Exception e) {
            LOG.error("处理建表事件异常", e);
        }
    }
    
    @Override
    public void onAlterTable(AlterTableEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        // 可以监控表结构变更
        // 这里暂时不做告警
        try {
            Table oldTable = event.getOldTable();
            Table newTable = event.getNewTable();
            
            LOG.debug("检测到表修改: {}.{}", newTable.getDbName(), newTable.getTableName());
            
        } catch (Exception e) {
            LOG.error("处理修改表事件异常", e);
        }
    }
    
    @Override
    public void onAddPartition(AddPartitionEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        // 可以监控分区增加
        // 这里暂时不做告警
    }
    
    @Override
    public void onDropPartition(DropPartitionEvent event) throws org.apache.hadoop.hive.metastore.api.MetaException {
        // 可以监控分区删除
        // 如果短时间内删除大量分区，可以告警
    }
    
    /**
     * 发送告警
     */
    private void sendAlert(AlertMessage alert) {
        if (executorService != null) {
            // 异步发送
            executorService.submit(() -> {
                try {
                    alertSender.send(alert);
                } catch (Exception e) {
                    LOG.error("发送告警失败", e);
                }
            });
        } else {
            // 同步发送
            try {
                alertSender.send(alert);
            } catch (Exception e) {
                LOG.error("发送告警失败", e);
            }
        }
    }
    
    /**
     * Hook销毁时清理资源
     */
    public void shutdown() {
        if (executorService != null) {
            LOG.info("关闭告警发送线程池");
            executorService.shutdown();
        }
    }
}

